package beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor;

import beautiful.butterfly.distributed_task_processing_engine.api.MultipleTaskDataItemCollectorAndExecuterAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.api.SingleTaskDataItemCollectorAndExecuterAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.api.TaskDataItemCollectorAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.TaskItemStringAndParameterString;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.Statistics;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.TaskProcessor;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskUserDefinition;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
public class WaitAllFinishTaskDataItemProcessor<TaskDataItem> implements ITaskDataItemProcessor, Runnable
{
    final ObjectLock objectLock = new ObjectLock();
    @Setter
    @Getter
    protected TaskProcessor taskProcessor;
    @Setter
    @Getter
    protected TaskDataItemCollectorAndExecuter<TaskDataItem> taskDataItemCollectorAndExecuter;
    @Setter
    @Getter
    protected List<TaskDataItem> taskDataItemList = new CopyOnWriteArrayList<TaskDataItem>();
    @Setter
    @Getter
    List<Thread> threadList = new CopyOnWriteArrayList<Thread>();
    @Setter
    @Getter
    TaskUserDefinition taskUserDefinition;
    @Setter
    @Getter
    boolean multipleTask = false;

    @Setter
    @Getter
    boolean stopSchedule = false;
    @Setter
    @Getter
    boolean isSleeping = false;
    @Setter
    @Getter
    Statistics statistics;


    public WaitAllFinishTaskDataItemProcessor(//
                                              TaskProcessor taskProcessor,//
                                              TaskDataItemCollectorAndExecuter<TaskDataItem> taskDataItemCollectorAndExecuter,//
                                              Statistics statistics) throws Exception//
    {
        this.taskProcessor = taskProcessor;
        this.taskUserDefinition = this.taskProcessor.getTaskUserDefinition();
        //
        this.taskDataItemCollectorAndExecuter = taskDataItemCollectorAndExecuter;
        if (this.taskDataItemCollectorAndExecuter instanceof SingleTaskDataItemCollectorAndExecuterAndExecuter<?>)
        {
            if (taskUserDefinition.getMaxExecuteNumber() > 1)
            {
                taskUserDefinition.setMaxExecuteNumber(1);//自动修正
            }
            multipleTask = false;
        } else
        {
            multipleTask = true;
        }
        if (taskUserDefinition.getFetchNumber() >= taskUserDefinition.getThreadNumber() * 10)
        {
            log.warn("参数设置不合理，系统性能不佳。【每次从数据库获取的数量fetchnum】 >= 【线程数量threadnum】 *【最少循环次数10】 ");
        }
        //
        this.statistics = statistics;
        //
        for (int i = 0; i < taskUserDefinition.getThreadNumber(); i++)
        {
            this.startThread(i);
        }
    }


    private void startThread(int index)
    {
        Thread thread = new Thread(this);//共享信息
        threadList.add(thread);
        String threadName = //
                this.taskProcessor.getTaskProcessorInfo().getTaskId() + "-"//
                        + this.taskProcessor.getCurrentThreadNo() + "-exe"//
                        + index;//
        thread.setName(threadName);
        thread.start();
    }

    //
    public void clearTaskDataItemList()
    {
        this.taskDataItemList.clear();
    }


    public boolean runningTaskItemListIsAllFinshed()
    {
        return this.taskDataItemList.size() == 0;
    }

    public boolean isSleeping()
    {
        return this.isSleeping;
    }

    /**
     * 需要注意的是，调度服务器从配置中心注销的工作，必须在所有线程退出的情况下才能做
     */
    public void setStopScheduleFlagAndClearTaskList() throws Exception
    {
        // 设置停止调度的标志,调度线程发现这个标志，执行完当前任务后，就退出调度
        this.stopSchedule = true;
        //清除所有未处理任务,但已经进入处理队列的，需要处理完毕
        this.taskDataItemList.clear();
    }


    //
    public void run()
    {
        try
        {
            long startTimeMillis = 0;
            while (true)
            {

                //
                this.objectLock.increaseThreadCount1();
                Object taskItem;
                while (true)
                {
                    //while (true) do{this}
                    if (this.stopSchedule)
                    {//停止队列调度
                        //
                        this.objectLock.reduceThreadCount1();
                        this.objectLock.objectNotifyAll();
                        //
                        synchronized (this.threadList)
                        {
                            this.threadList.remove(Thread.currentThread());
                            if (this.threadList.size() == 0)
                            {
                                this.taskProcessor.unRegisterTaskProcessor();
                            }//all over
                        }
                        return;
                    }
                    //!setStopScheduleFlagAndClearTaskList
                    //加载调度任务
                    //load jvm_instance_info_and_task_processor_allocation_strategy from here
                    if (this.multipleTask == false)
                    {
                        taskItem = this.getTaskDataItem();
                    } else
                    {
                        taskItem = this.getTaskDataItems();
                    }

                    if (taskItem == null)
                    {
                        break;
                    }

                    try
                    {
                        startTimeMillis = taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
                        if (!this.multipleTask)//single_task
                        {
                            if (((SingleTaskDataItemCollectorAndExecuterAndExecuter) this.taskDataItemCollectorAndExecuter).dealWith(taskItem) == true)
                            {
                                addTaskSuccessNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                        - startTimeMillis);
                            } else
                            {
                                addFailNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                        - startTimeMillis);
                            }
                        } else
                        {//multiple_task
                            if (((MultipleTaskDataItemCollectorAndExecuterAndExecuter) this.taskDataItemCollectorAndExecuter)
                                    .dealWith((Object[]) taskItem) == true)
                            {
                                addTaskSuccessNum(((Object[]) taskItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                        - startTimeMillis);
                            } else
                            {
                                addFailNum(((Object[]) taskItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                        - startTimeMillis);
                            }
                        }
                    } catch (Throwable ex)
                    {
                        if (!this.multipleTask)
                        {
                            addFailNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis() - startTimeMillis);
                        } else
                        {
                            addFailNum(((Object[]) taskItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                    - startTimeMillis);
                        }
                        log.warn("TaskDataItem :" + taskItem + " ", ex);
                    }
                    //整个处理器进行报告,单个任务项不行
                    getTaskProcessor().getTaskAndTaskProcessorAndTaskDataItemManage().reportTaskProcessorDealInfo(getTaskProcessor().getTaskProcessorInfo(), this.statistics);
                }//end while
                log.info("队列中的数据已经运行完成，当前运行线程数量:{}", this.objectLock.getThreadCount());
                //has set(this.objectLock.increaseThreadCount1();)
                if (this.objectLock.reduceThreadCount1IfThreadCountNotEqual1AndReturnTheCurrentThreadIsNotLastThread())
                {
                    this.objectLock.objectWait();
                } else
                {//current thread is the last thread
                    log.info("任务执行完成，再次加载数据");
                    Thread.currentThread().sleep(100);
                    startTimeMillis = this.taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
// 装载数据
                    int taskSize = this.fetchTaskDataItems();
                    if (taskSize > 0)
                    {
                        this.objectLock.objectNotifyAll();
                    } else
                    {
                        //判断当没有数据的是否，是否需要退出调度
                        if (this.stopSchedule == false && this.taskProcessor.continueWhenHasData())
                        {
                            this.isSleeping = true;
                            Thread.currentThread().sleep(this.taskProcessor.getTaskUserDefinition().getSleepTimeWhenNotHasData());
                            this.isSleeping = false;

                        } else
                        {
                            this.objectLock.objectNotifyAll();
                        }
                    }
                    this.objectLock.reduceThreadCount1();
                }
            }
        } catch (Throwable e)
        {
            log.error(e.getMessage(), e);
        }
    }

    public synchronized Object getTaskDataItem()
    {
        if (this.taskDataItemList.size() == 0)
        {
            return null;
        } else
        {
            return this.taskDataItemList.remove(0);// 按正序处理
        }
    }

    public synchronized Object[] getTaskDataItems()
    {
        if (this.taskDataItemList.size() == 0)//0
        {
            return null;
        }

        int minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskDataItemList.size();
        if (taskUserDefinition.getMaxExecuteNumber() == 0)
        {//0
            //不做限制
        } else
        {
            minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskDataItemList.size() > taskUserDefinition.getMaxExecuteNumber() ? taskUserDefinition.getMaxExecuteNumber() : taskDataItemList.size();//min
        }
        Object[] taskList = (Object[]) Array.newInstance(this.taskDataItemList.get(0).getClass(), minSizeBetweenTaskListSizeAndMaxExecuteNumber);
        for (int i = 0; i < minSizeBetweenTaskListSizeAndMaxExecuteNumber; i++)
        {
            taskList[i] = this.taskDataItemList.remove(0);// 按正序处理
        }
        return taskList;

    }

    /**
     * 获取任务  为什么这种方式不会出现重复的问题:因为采用同步控制的方式,只有一个线程去获取数据
     *
     * @return 任务数量
     */
    protected int fetchTaskDataItems()
    {
        try
        {   //在每次数据处理完毕后休眠固定的时间
            if (this.taskUserDefinition.getSleepTimeIntervalBeforeFetchTasks() > 0)
            {

                this.isSleeping = true;
                Thread.sleep(taskUserDefinition.getSleepTimeIntervalBeforeFetchTasks());
                this.isSleeping = false;

            }
            List<TaskItemStringAndParameterString> taskItemStringAndParameterStringList = this.taskProcessor.waiterunningTaskItemListIsAllFinshedAndSynchronizedGetTaskItemStringAndParameterStringList();
            if (taskItemStringAndParameterStringList.size() > 0)
            {
                //底层的对象不可以被共享
                List<TaskItemStringAndParameterString> newTaskItemStringAndParameterStringList = new ArrayList<TaskItemStringAndParameterString>();
                synchronized (taskItemStringAndParameterStringList)
                {
                    for (TaskItemStringAndParameterString taskItemStringAndParameterString : taskItemStringAndParameterStringList)
                    {
                        newTaskItemStringAndParameterStringList.add(taskItemStringAndParameterString);
                    }
                }
                List<TaskDataItem> taskDataItemList = this.taskDataItemCollectorAndExecuter.getTaskDataItemList(//
                        taskUserDefinition.getTaskParameter(),//
                        taskUserDefinition.getFetchNumber(),//
                        this.taskProcessor.getTaskItemCount(), //
                        newTaskItemStringAndParameterStringList //
                );//
                taskProcessor.getTaskProcessorInfo().setLastFetchDataTimestamp(new Timestamp(taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()));
                if (taskDataItemList != null)
                {
                    this.taskDataItemList.addAll(taskDataItemList);
                }
            }
            addTaskFetchNum(taskDataItemList.size());
            return this.taskDataItemList.size();
        } catch (Throwable ex)
        {
            log.error("Get tasks error.", ex);
        }
        return 0;
    }

    //
    public void addTaskFetchNum(long num)
    {

        this.statistics.addFetchDataCount(1);
        this.statistics.addTaskFetchNum(num);
    }

    public void addTaskSuccessNum(long num, long spendTime)
    {
        this.statistics.addDealDataSucess(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime)
    {
        this.statistics.addDealDataFail(num);
        this.statistics.addDealSpendTime(spendTime);
    }
}
